package com.zyx.flinkdemo.sql.utils;

/**
 * @author zyx
 * @since 2021/5/28 23:13
 * desc: 获取Flink连接拼接字符串的工具类
 */
public class ConnectUtils {

    public static String getCreateKafkaSourceTableSql(String kafkaTopic,
                                                      String kafkaServer,
                                                      String kafkaConsumerGroup,
                                                      String kafkaFormat,
                                                      String scanStartUpMode,
                                                      String createKafkaSourceSql) {
        return createKafkaSourceSql + " WITH ( "
                +" 'connector' = 'kafka' ,"
                + " 'topic' = '" + kafkaTopic + "',"
                + " 'properties.bootstrap.servers' = '" + kafkaServer + "',"
                + " 'properties.group.id' = '" + kafkaConsumerGroup + "',"
                + " 'format' = '" + kafkaFormat + "',"
                + " 'scan.startup.mode' = '" + scanStartUpMode + "',"
                + " 'json.fail-on-missing-field' = 'false',"
                + " 'json.ignore-parse-errors' = 'true' )";
    }

    public static String getNormalKafkaSourceConnect(String tblName,
                                         String tblFields,
                                         String topic,
                                         String kafkaServer,
                                         String consumerGroup,
                                         String startUpMode,
                                         String format) {
        return String.format("CREATE TABLE %s (\n" +
                "%s" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '%s',\n" +
                "  'properties.bootstrap.servers' = '%s',\n" +
                "  'properties.group.id' = '%s',\n" +
                "  'scan.startup.mode' = '%s',\n" +
                "  'format' = '%s'\n" +
                ")", tblName, tblFields, topic, kafkaServer, consumerGroup, startUpMode, format);
    }

    public static String getNormalKafkaSinkConnect(String tblName,
                                               String tblFields,
                                               String topic,
                                               String kafkaServer,
                                               String startUpMode,
                                               String format) {
        return String.format("CREATE TABLE %s (\n" +
                "%s" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '%s',\n" +
                "  'properties.bootstrap.servers' = '%s',\n" +
                "  'scan.startup.mode' = '%s',\n" +
                "  'format' = '%s'\n" +
                ")", tblName, tblFields, topic, kafkaServer, startUpMode, format);
    }

    public static String getNormalMysqlSinkConnect(String tblName,
                                                   String fields,
                                                   String host,
                                                   String port,
                                                   String mysqlDb,
                                                   String mysqlTbl,
                                                   String username,
                                                   String password) {
        return String.format("CREATE TABLE %s (\n" +
                "%s\n" +
                ") WITH (\n" +
                "   'connector' = 'jdbc',\n" +
                "   'url' = 'jdbc:mysql://%s:%s/%s?characterEncoding=utf8&useSSL=false',\n" +
                "   'table-name' = '%s',\n" +
                "   'username' = '%s',\n" +
                "   'password' = '%s'\n" +
                ")", tblName, fields, host, port, mysqlDb, mysqlTbl, username, password);
    }
}
